//package com.teamsun.mqtt.service;
//
//import io.netty.buffer.ByteBuf;
//import io.netty.util.ReferenceCountUtil;
//
//import java.util.ArrayList;
//import java.util.List;
//import java.util.concurrent.BlockingQueue;
//import java.util.concurrent.ConcurrentHashMap;
//import java.util.concurrent.ConcurrentMap;
//import java.util.concurrent.DelayQueue;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.LinkedBlockingQueue;
//import java.util.stream.Stream;
//
//import org.apache.log4j.Logger;
//import org.springframework.beans.factory.annotation.Autowired;
//
//import com.teamsun.mqtt.entity.SendMsg;
//import com.teamsun.mqtt.entity.vo.DelaySnedMsg;
//import com.teamsun.mqtt.entity.vo.DelayWillRemoveMsg;
//import com.teamsun.mqtt.entity.vo.SendableMsg;
//
///**
// * 重发机制服务
// * 
// * @author tzj
// * 
// */
//public class ReSendService$ {
//
//	Logger logger = Logger.getLogger(getClass());
//	/**
//	 * 延迟队列
//	 */
//	DelayQueue<DelaySnedMsg> delayQueue = new DelayQueue<DelaySnedMsg>();
//
//	/**
//	 * 用于映射客户端标识和延迟对象
//	 */
//	ConcurrentMap<String, DelaySnedMsg> concurrentMap = new ConcurrentHashMap<String, DelaySnedMsg>();
//
//	/**
//	 * 客户端与 “客户端 消息”这样的组的映射
//	 */
//	ConcurrentHashMap<String, BlockingQueue<String>> clentid2clientAndMsgs = new ConcurrentHashMap<String, BlockingQueue<String>>();
//
//	
//	
//	@Autowired
//	MessagePushService messagePushService;
//	
//	/**
//	 * 数据库操作线程组
//	 */
//	ExecutorService dboptgroup;
//
//	 int  resendtimes=3;
//
//	public ReSendService$() {
//		//start();
//	}
//
//	
//	/**
//	 * 提交一个 延迟发送对象到 延迟队列
//	 * 
//	 * 至于延迟对象的 buff 首先在map里面找，如果能找到就标记保留+1 标记的时候有可能异常 在异常里面新建buff给延迟对象
//	 * 
//	 * @param delaySnedMsg
//	 */
//
//	public void submitDelayMsg(DelaySnedMsg delaySnedMsg) {
//
//		SendableMsg sendableMsg = delaySnedMsg.getSendableMsg();
//		String clientid = sendableMsg.getRecvclientid();
//		String  cAndmid=clientid+":"+sendableMsg.getMessageid();
//		ByteBuf willSendBytebuff = sendableMsg.getMsgContent().retainedDuplicate();
//		sendableMsg.setDupTimes(sendableMsg.getDupTimes()+1);
//		
//		if (willSendBytebuff.refCnt() <= 0) {
//			logger.error("消息已经被释放,添加延迟队列失败");
//			return;
//		}
//
//		delayQueue.offer(delaySnedMsg);
//		DelaySnedMsg delaySnedMsglod = concurrentMap.put(cAndmid, delaySnedMsg);
//		BlockingQueue<String> blockingQueue = null;
//		if (!clentid2clientAndMsgs.containsKey(clientid)) {
//			blockingQueue = new LinkedBlockingQueue<String>();
//			clentid2clientAndMsgs.put(clientid, blockingQueue);
//		} else
//			blockingQueue = clentid2clientAndMsgs.get(clientid);
//
//		if (blockingQueue != null && delaySnedMsglod == null)
//			blockingQueue.offer(cAndmid);
//
//	}
//
//	/**
//	 * 启动一个线程重新发送消息给客户端
//	 */
//	public void start() {
//
//		new Thread(()->{
//			
//			while (true) {
//				DelaySnedMsg delaySnedMsg = null;
//				SendableMsg sendableMsg = null;
//				try {
//					delaySnedMsg = delayQueue.take();
//					if(delaySnedMsg.isHasSend())
//						continue;
//					sendableMsg = delaySnedMsg.getSendableMsg();
//				} catch (InterruptedException e) {
//					logger.debug(e);
//				}
//
//				if (delaySnedMsg != null && sendableMsg != null) {
//
//					if (sendableMsg.getDupTimes() >= resendtimes) {// 3
//						// 应该可配置
//						SendMsg unsceesssMsg = new SendMsg(sendableMsg
//								.getRecvclientid(), sendableMsg
//								.getMessageid(),
//								messagePushService.MSG_SENDUNSUCCESS,
//								sendableMsg.getSendclientid());
//						messagePushService.delayOCalQueue.offer(new DelayWillRemoveMsg(unsceesssMsg));
//						ReferenceCountUtil.release(sendableMsg.getMsgContent());
//						logger.debug(sendableMsg.getSendclientid()+"重发给"+sendableMsg.getRecvclientid()+"三次不成功，加入成功队列");
//						sendableMsg=null;
//					}
//
//					else {
//					   if(sendableMsg.getMsgContent().refCnt()<=0){
//						   logger.warn("消息已经被释放,延迟发送失败");
//					   }else{
//						   logger.debug( "第"+sendableMsg.getDupTimes()+"次给"+sendableMsg.getSendclientid()+"重新发送给"+sendableMsg.getRecvclientid()+
//								   "消息主题:"+sendableMsg.getTopname());
//						
//							messagePushService.sendMsg2Client(sendableMsg);
//					   }
//					}
//
//				}
//
//			}
//			
//		}).start();;
//		
//	}
//
//	/**
//	 * 返回演出消息对象
//	 * 
//	 * @param clientid
//	 * @param msgid
//	 * @return
//	 */
//	public DelaySnedMsg getDelaySnedMsg(String clientid, Integer msgid) {
//		String cAndmid = clientid + ":" + msgid;
//		if (concurrentMap.containsKey(cAndmid)) {
//			return concurrentMap.get(cAndmid);
//		}
//		return null;
//	}
//
//	/**
//	 * 根据客户端标识移除 延迟队列消息 首先根据客户端标识找到 客户端+消息 队列 遍历 客户端+消息 队列 得到每一个延迟消息 从 延迟队列移除它
//	 * 客户端+消息 队列 根据客户端标识移除键 clentid2clientAndMsgs
//	 * 
//	 * @param clentid
//	 * @param recvsuccess
//	 *            是否是成功接收了 客户端标识
//	 * @return 成功与否
//	 */
//	public boolean remmoveFromClientId(final String clentid, boolean recvsuccess) {
//
//		
//		//logger.debug("删除延迟队列客户端" + clentid + "的消息");
//		
//	
//		if (clentid2clientAndMsgs.containsKey(clentid)) {
//			BlockingQueue<String> blockingQueue = clentid2clientAndMsgs
//					.get(clentid);
//			List<String> strings = new ArrayList<String>();
//			blockingQueue.drainTo(strings);
//			
//			
//			Stream<String> stream=strings.parallelStream();
//			
//			stream.forEach((String string)->{
//				
//				DelaySnedMsg delaySnedMsg = concurrentMap.get(string);
//
//				if (delaySnedMsg != null) {
//
//					if (!recvsuccess) {
//						SendableMsg sendableMsg = delaySnedMsg.getSendableMsg();
//						SendMsg msg = new SendMsg(
//								sendableMsg.getRecvclientid(),
//								sendableMsg.getMessageid(),
//								messagePushService.MSG_SENDUNSUCCESS,
//								sendableMsg.getSendclientid());
//						messagePushService.delayOCalQueue.offer(new DelayWillRemoveMsg(msg));
//					}
//
//					delayQueue.remove(delaySnedMsg);
//				}
//				concurrentMap.remove(string);
//			});
//			
//			stream.close();
//			blockingQueue = null;
//			clentid2clientAndMsgs.remove(clentid);
//
//		}
//
//		return false;
//	}
//
//	public void setResendtimes(int resendtimes) {
//		this.resendtimes = resendtimes;
//	}
//
//
//
//	public ExecutorService getDboptgroup() {
//		return dboptgroup;
//	}
//
//	public void setDboptgroup(ExecutorService dboptgroup) {
//		this.dboptgroup = dboptgroup;
//	}
//	
//}
